package com.exm.collectcodenew.kafka.producer.customPartitioner;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import java.util.List;
import java.util.Map;
import java.util.Random;
public class AuditPartitioner implements Partitioner {
    private Random random;
    @Override
    public void configure(Map<String, ?> map) {
        //该方法实现必要资源的初始化工作
        random= new Random();
    }
    @Override
    public int partition(String topic, Object keyObj, byte[] keyBytes, Object valueObj, byte[] valueBytes, Cluster cluster) {
        String key=(String)keyObj;
        //从集群元数据中把属于该topic的所有分区信息都读取出供分区策略使用
        List<PartitionInfo> partitionInfoList = cluster.availablePartitionsForTopic(topic);
        int partitionCount =partitionInfoList.size();
        int auditPartition=partitionCount-1;
        return key == null|| key.isEmpty()|| !key.contains ("audit")?random.nextInt(partitionCount-1):auditPartition;
    }
    @Override
    public void close() {
    //该方法实现必要资源的清理工作
    }
}
